package com.xxl.job.executor.service.jobhandler;

import com.xxl.job.core.context.XxlJobContext;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.executor.service.loghandler.AnalysisStatistics;
import com.xxl.job.executor.service.loghandler.LogStatistics;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.concurrent.FutureTask;

/**
 * XxlJob开发示例（Bean模式）
 * <p>
 * 开发步骤：
 * 1、任务开发：在Spring Bean实例中，开发Job方法；
 * 2、注解配置：为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")"，注解value值对应的是调度中心新建任务的JobHandler属性的值。
 * 3、执行日志：需要通过 "XxlJobHelper.log" 打印执行日志；
 * 4、任务结果：默认任务结果为 "成功" 状态，不需要主动设置；如有诉求，比如设置任务结果为失败，可以通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置任务结果；
 *
 * @author xuxueli 2019-12-11 21:52:51
 */
@Component
public class DataxJob {
    private static Logger logger = LoggerFactory.getLogger(DataxJob.class);
    public static final String TASK_START_TIME_SUFFIX = "任务启动时刻";
    public static final String TASK_END_TIME_SUFFIX = "任务结束时刻";
    public static final String TASK_TOTAL_TIME_SUFFIX = "任务总计耗时";
    public static final String TASK_AVERAGE_FLOW_SUFFIX = "任务平均流量";
    public static final String TASK_RECORD_WRITING_SPEED_SUFFIX = "记录写入速度";
    public static final String TASK_RECORD_READER_NUM_SUFFIX = "读出记录总数";
    public static final String TASK_RECORD_WRITING_NUM_SUFFIX = "读写失败总数";
    @Value("${datax.path.bin}")
    private String dataxPathBin;

    @XxlJob("DataxJobHandler")
    public void DataxJobHandler() throws Exception {
        int exitValue = -1;
        Thread errThread = null;

        String json = XxlJobHelper.getJobParam();
        String fileName = "temp" + json.hashCode() + ".json";
        File file = new File(dataxPathBin + "/" + fileName);
        if (file.exists()) {
            file.delete();
        }
        file.createNewFile();

        Writer write = new OutputStreamWriter(new FileOutputStream(file), "UTF-8");
        write.write(json);
        write.flush();
        write.close();

        LogStatistics logStatistics = null;

        try {
            final Process process = Runtime.getRuntime().exec("python datax.py " + fileName, null, new File(dataxPathBin));
            // log-thread
            Thread futureThread = null;
            FutureTask<LogStatistics> futureTask = new FutureTask<>(() -> AnalysisStatistics.analysisStatisticsLog(new BufferedInputStream(process.getInputStream())));
            futureThread = new Thread(futureTask);
            futureThread.start();

            errThread = new Thread(() -> {
                try {
                    AnalysisStatistics.analysisStatisticsLog(new BufferedInputStream(process.getErrorStream()));
                } catch (IOException e) {
                    XxlJobHelper.log(e);
                }
            });

            logStatistics = futureTask.get();
            errThread.start();
            // process-wait
            exitValue = process.waitFor();      // exit code: 0=success, 1=error
            // log-thread join
            errThread.join();
        } catch (Exception e) {
            XxlJobHelper.log(e);
        } finally {
            if (errThread != null && errThread.isAlive()) {
                errThread.interrupt();
            }
            //  删除临时文件
            if (file.exists()) {
                file.delete();
            }
        }
        if (exitValue == 0) {
            XxlJobContext.getXxlJobContext().setHandleMsg(logStatistics.toString());
        } else {
            XxlJobContext.getXxlJobContext().setHandleCode(500);
            XxlJobContext.getXxlJobContext().setHandleMsg("任务执行失败，请查看日志排查错误");
        }
    }


//
//    private String translateLog(String line) {
//        if (line.contains("ERROR")) {
//return line;
//        }else if (line.contains())
//    }
}
